package org.apache.hadoop.hdfs.shortcircuit;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.hdfs.net.DomainPeer;
import org.apache.hadoop.hdfs.shortcircuit.DfsClientShmManager.EndpointShmManager;
import org.apache.hadoop.net.unix.DomainSocket;
import org.apache.hadoop.net.unix.DomainSocketWatcher;

import com.google.common.base.Preconditions;

/**
 * DfsClientShm类用于在DFSClient侧抽象共享内存，这个类是ShortCircuitShm的子类。要注意这个类实现了DomainSocketWatcher.Handler接口，
 *
 * 当DFSClient和Datanode进程通信的domainSocket对象关闭时，这个DfsClientShm会被标识为stale状态。
 * stale状态的DfsClientShm中保存的所有Slot对象会被设置为无效，同时Slot对象对应数据块的ShortCircuitReplica对象会被同步设置为stale状态，
 * 也就不能再接收新的读请求以及缓存请求了， 但是当前正在进行的读操作不会受到影响。
 * 当一个Slot对象对应的ShortCircuitReplica上所有的读取操作完成后，ShortCircuitCache就会关闭这个ShortCircuitReplica，并且释放对应的槽位。
 * 如果共享内存段中所有的槽位都已经释放，就可以释放这段共享内存了。
 * 之所以这样设计，是因为如果共享内存匹配的domainSocket关闭后，Datanode侧的共享内存状态和DFSClient侧的共享内存不能确保是同步更新的，
 * 所以需要引入DomainSocketWatcher.Handler接口处理domainSocket关闭的情况。
 */
public class DfsClientShm extends ShortCircuitShm implements DomainSocketWatcher.Handler {
  /**
   * The EndpointShmManager associated with this shared memory segment.
   */
  private final EndpointShmManager manager;

  /**
   * The UNIX domain socket associated with this DfsClientShm.
   * We rely on the DomainSocketWatcher to close the socket associated with
   * this DomainPeer when necessary.
   */
  private final DomainPeer peer;

  /**
   * True if this shared memory segment has lost its connection to the
   * DataNode.
   *
   * {@link DfsClientShm#handle} sets this to true.
   */
  private boolean disconnected = false;

  DfsClientShm(ShmId shmId, FileInputStream stream, EndpointShmManager manager, DomainPeer peer) throws IOException {
    super(shmId, stream);
    this.manager = manager;
    this.peer = peer;
  }

  public EndpointShmManager getEndpointShmManager() {
    return manager;
  }

  public DomainPeer getPeer() {
    return peer;
  }

  /**
   * Determine if the shared memory segment is disconnected from the DataNode.
   *
   * This must be called with the DfsClientShmManager lock held.
   *
   * @return   True if the shared memory segment is stale.
   */
  public synchronized boolean isDisconnected() {
    return disconnected;
  }

  /**
   * Handle the closure of the UNIX domain socket associated with this shared
   * memory segment by marking this segment as stale.
   *
   * If there are no slots associated with this shared memory segment, it will
   * be freed immediately in this function.
   */
  @Override
  public boolean handle(DomainSocket sock) {
    manager.unregisterShm(getShmId());
    synchronized (this) {
      Preconditions.checkState(!disconnected);
      // 将共享内存段设置为stable状态
      disconnected = true;
      boolean hadSlots = false;
      // 将共享内存中的所有slot都设置为无效
      for (Iterator<Slot> iter = slotIterator(); iter.hasNext(); ) {
        Slot slot = iter.next();
        slot.makeInvalid();
        hadSlots = true;
      }
      if (!hadSlots) {
        // 如果共享内存中的所有槽位都被释放了，则调用free()方法释放共享内存
        free();
      }
    }
    return true;
  }
}
